package com.example.rpc;

import cn.hutool.core.lang.UUID;
import com.example.impl.MyCat;
import com.example.interf.Cat;
import com.example.protocol.PackMsg;
import com.example.protocol.ReqBody;
import com.example.protocol.ReqHeader;
import com.example.protocol.RespBody;
import com.example.serialization.SeriaUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.junit.Test;

import java.io.IOException;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

public class MyRPC {

    @Test
    public void server() {

        MyCat cat = new MyCat();
        Dispatcher.register(Cat.class.getName(), cat);


        NioEventLoopGroup eventExecutors = new NioEventLoopGroup(10);
        NioEventLoopGroup worker = eventExecutors;

        ServerBootstrap bs = new ServerBootstrap();
        try {
            bs.group(eventExecutors, worker)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<NioSocketChannel>() {
                        @Override
                        protected void initChannel(NioSocketChannel ch) throws Exception {
                            System.out.println("server rec client port: " + ch.remoteAddress().getPort());
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new ServerDecoder());
                            pipeline.addLast(new MyAcceptHandler());
                        }
                    }).bind(new InetSocketAddress("localhost", 9090))
                    .sync()
                    .channel()
                    .closeFuture()
                    .sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    /**
     * 模拟客户端
     */
    @Test
    public void app() {

        new Thread(this::server).start();

        System.out.println("server start...");

        AtomicInteger ind = new AtomicInteger(0);
        Thread[] threads = new Thread[20];
        for (int i = 0; i < 20; i++) {
            threads[i] = new Thread(() -> {
                Cat cat = proxyGet(Cat.class);
                String arg = "hello cat " + ind.incrementAndGet();
                String res = cat.say(arg);
                System.out.println("callback " + res + " src arg: " + arg);
            });
        }
        Arrays.stream(threads).forEach(Thread::start);


        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }


    }

    private <T> T proxyGet(Class<T> interfaceClazz) {

        ClassLoader classLoader = interfaceClazz.getClassLoader();
        Class<?>[] interfaces = {interfaceClazz};

        T o = (T) Proxy.newProxyInstance(classLoader, interfaces, new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                // 如何设计consumer对provider的远程调用

                // 通信协议：请求体 包含：哪个类，哪个方法，方法参数类型，方法参数值
                String methodName = method.getName();
                Class<?>[] parameterTypes = method.getParameterTypes();

                ReqBody reqBody = new ReqBody();
                reqBody.setArgs(args);
                reqBody.setMethodName(methodName);
                reqBody.setName(interfaceClazz.getName());
                reqBody.setParameterTypes(parameterTypes);
                byte[] bodybytes = SeriaUtil.encode(reqBody);

                // 通信协议：请求头 包含：状态标志 int 32位状态，请求体长度，请求ID
                ReqHeader reqHeader = new ReqHeader();
                reqHeader.setDataLen(bodybytes.length);
                reqHeader.setRequestID(Math.abs(UUID.fastUUID().getLeastSignificantBits()));
                reqHeader.setFlag(0x14141414);
                byte[] headerbytes = SeriaUtil.encode(reqHeader);

                CompletableFuture<String> cmf = new CompletableFuture<>();
                ResponseHandler.addCallback(reqHeader.getRequestID(), cmf);

                // 连接池
                ByteBuf buf = ByteBufAllocator.DEFAULT.directBuffer(headerbytes.length + bodybytes.length);
                buf.writeBytes(headerbytes);
                buf.writeBytes(bodybytes);
                ClientFactory.factory
                        .getClient(new InetSocketAddress("localhost", 9090))
                        .writeAndFlush(buf)
                        .sync();


                // 等待返回的机制，方法调用是同步的，而使用netty进行网络通信是基于event 驱动，异步返回

                return cmf.get();
            }
        });

        return o;
    }

    private class MyAcceptHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

            PackMsg packMsg = (PackMsg) msg;

            String ioThreadName = Thread.currentThread().getName();

            ctx.executor().parent().next().execute(() -> {
//            ctx.executor().execute(() -> {

//                String execThreadName = Thread.currentThread().getName();
//
//                String s = "io thread: " + ioThreadName + "  exec thread: " + execThreadName + "  req arg = " + packMsg.getReqBody().getArgs()[0];

                String clazz = packMsg.getReqBody().getName();
                String methodName = packMsg.getReqBody().getMethodName();
                Class<?>[] parameterTypes = packMsg.getReqBody().getParameterTypes();
                Object[] args = packMsg.getReqBody().getArgs();

                Object impl = Dispatcher.get(clazz);

                Class<?> implClass = impl.getClass();
                Object res = null;
                try {
                    Method method = implClass.getMethod(methodName, parameterTypes);
                    res = method.invoke(impl, args);

                } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
                    e.printStackTrace();
                }

                RespBody respBody = new RespBody();
                respBody.setMsg((String) res);
                byte[] byteBody = SeriaUtil.encode(respBody);

                ReqHeader header = packMsg.getHeader();
                header.setFlag(0x14141424);
                header.setDataLen(byteBody.length);
                byte[] byteHeader = SeriaUtil.encode(header);

                ByteBuf buf = ByteBufAllocator.DEFAULT.directBuffer(byteHeader.length + byteBody.length);
                buf.writeBytes(byteHeader);
                buf.writeBytes(byteBody);
                ctx.writeAndFlush(buf);
            });


        }
    }
}
